﻿using InCom.SyncStatus;
using InCom.Utilities;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace InCom.Core.TerminalSend.MappedFile
{
    /// <summary>
    /// 发送时实数据到磁盘中
    /// </summary>
    public class RealSendBytesToDisk
    {
        /// <summary>
        ///需要数据持久化功能的节点.才使用此对像
        /// </summary>
        internal RealDataSendPersistence persistence = new RealDataSendPersistence();

        /// <summary>
        /// 发送线程任务集合
        /// </summary>
        internal ConcurrentDictionary<string, SendTagsBuffer> Sendtasks = new ConcurrentDictionary<string, SendTagsBuffer>();

        /// <summary>
        /// 开始记数时间
        /// </summary>
        internal DateTime startTime = DateTime.Now;

        /// <summary>
        ///  每秒写入记录总数 xxx/s
        /// </summary>
        internal long WritePackCount = 0;

        /// <summary>
        /// 节点对像
        /// </summary>
        private List<string> TagNames = new List<string>();

        /// <summary>
        /// 写入数据,服务器上实际测试不到2000条记录每秒
        /// </summary>
        /// <param name="data">需要发送的数据</param>
        /// <param name="syncPoint">同步节点</param>
        public void Write(byte[] data, SyncPoint syncPoint)
        {
            //需要写入的数据
            byte[] writeData = new byte[data.Length + 4];

            //数据复制
            //4字节长度
            Buffer.BlockCopy(BitConverter.GetBytes(data.Length), 0, writeData, 0, 4);
            //协议内容
            Buffer.BlockCopy(data, 0, writeData, 4, data.Length);


            string key = persistence.GetKey(syncPoint.TagName);
            SendTagsBuffer buf = null;
            //获取Key
            Sendtasks.TryGetValue(key, out buf);
            //看看是否包含指定的节点项
            if (buf != null)
            {
                //如果没有找到对应 的写入持久化对像
                if (buf.PerStatus == null || buf.PerStatus.FileReadWrite == null || !buf.PerStatus.FileReadWrite.CanWriteRead)
                {
                    //写入不同的数据
                    persistence.Write(writeData, key, syncPoint.TagName, true);
                    //获取持久化数据对像
                    buf.PerStatus = persistence.GetPersistenceStatus(key);
                }
                else
                {
                    //写入不同的数据
                    buf.PerStatus.FileReadWrite.Write(writeData);
                }
                buf.WriteCount++;
            }
            else
            {
                buf = new SendTagsBuffer()
                {
                    TagKey = key,
                    StartTime = DateTime.Now,
                    CanRun = true,
                    SyncPoint = syncPoint
                };
                buf.Task = new TaskEx(Cycle, buf, new System.Threading.CancellationTokenSource());

                //写入不同的数据
                persistence.Write(writeData, syncPoint.TagName);
                lock (Sendtasks)
                {
                    Sendtasks.TryAdd(key, buf);
                    //获取持久化数据对像
                    buf.PerStatus = persistence.GetPersistenceStatus(key);
                }

                if (syncPoint.CanUser)
                {
                    //开始发送线程
                    buf.Task.Start();
                }

                //写入记录增加
                buf.WriteCount++;
            }
            //统计记数
            System.Threading.Interlocked.Add(ref WritePackCount, 1);
        }

        /// <summary>
        /// 终端注册后,启动发送实时数据线程
        /// </summary>
        /// <param name="syncPoint">同步节点</param>
        public void RegistTerminal(SyncPoint syncPoint)
        {
            if (syncPoint.CanUser)
            {
                //获取key
                string key = persistence.GetKey(syncPoint.TagName);
                if (Sendtasks.ContainsKey(key))
                {
                    var buf = Sendtasks[key];
                    //判断当前状态,是否需要重新创建线程
                    //重新开始线程
                    if (buf.CanRun == false && buf.Task != null && buf.Task.Status != System.Threading.Tasks.TaskStatus.Running)
                    {
                        //关闭线程
                        buf.Task.CanToKen.Cancel();
                        //再创建一个新线程
                        buf.Task = new TaskEx(Cycle, buf, new System.Threading.CancellationTokenSource());
                        buf.CanRun = true;
                        buf.Task.Start();

                    }
                    else
                    {
                        //节点可运行
                        if (buf.CanRun == true && buf.Task.Status == System.Threading.Tasks.TaskStatus.Created)
                        {
                            buf.Task.Start();
                            return;
                        }
                    }

                }
            }
        }


        #region 循环发送
        /// <summary>
        /// 循环
        /// </summary>
        protected void Cycle(object obj)
        {
            SendTagsBuffer tag = obj as SendTagsBuffer;
            while (tag.CanRun)
            {
                try
                {
                    SendToPoint(tag);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex);
                    System.Diagnostics.Trace.WriteLine(ex);
                }
            }
        }





        /// <summary>
        /// 发送数据到节点
        /// </summary>
        protected void SendToPoint(SendTagsBuffer per)
        {
            //是否发送最新
            bool SendNew = !(per.SyncPoint.SyncPointType == SyncPointType.RealDataCalculate);

            byte[] data = null;
            #region 节点不可用不进行计算,退出发送线程
            //不能使用直接跳过
            if (!per.SyncPoint.CanUser)
            {
                if (per.CanRun)
                {
                    //设置运行完成
                    per.CanRun = false;
                    //停止线程
                    per.Task.CanToKen.Cancel();
                    lock (Sendtasks)
                    {
                        SendTagsBuffer delobj = null;
                        Sendtasks.TryRemove(per.TagKey, out delobj);
                        //删除集合对像
                        if (delobj != null)
                        {
                            if (per.SendCount > 0)
                            {
                                per.SendCount = -1;
                                System.Threading.Thread.Sleep(10);
                                Console.WriteLine("将线程移除API发送队列,标识" + per.SyncPoint.TagName);
                            }
                        }
                    }
                    //不进行资源释放,后台会自动回收处理释放资源
                    //var obj = persistence.GetPersistenceStatus(per.TagKey);
                    //if (obj == null || obj.FileReadWrite == null)
                    //{
                    //    return;
                    //}
                    ////释放资源
                    //obj.FileReadWrite.Dispose();
                    //obj.FileReadWrite = null;
                }
                System.Threading.Thread.Sleep(200);
                return;
            }
            #endregion

            //获取上一次节点
            var obj = persistence.GetPersistenceStatus(per.TagKey);
            if (obj == null || obj.FileReadWrite == null)
            {
                System.Threading.Thread.Sleep(200);
                return;
            }
            //当前读取字节长度
            int curReadCount = 0;
            //单次最大读取长度
            int maxReadlen = 1024 * 120;
            //可以发送出去的数据
            List<byte[]> sendArrs = new List<byte[]>();
            //读取的有效数据长度
            int datavalidLen = 0;
            //判断数据长度,当可读取数据长度超过258KB时一次读取到只剩下10KB数据
            for (; obj.FileReadWrite.CurrentRead < obj.FileReadWrite.CureentWrite - 1024 * 10; )
            {
                #region 以前读取一包数据的处理方式
                int datalen = 0;
                //正常可以使用节点
                var dataLenBytes = persistence.Read(per.TagKey, true, 4);
                //增加当前读取长度
                curReadCount += dataLenBytes.Length;
                if (dataLenBytes.Length == 4)
                {
                    datalen = BitConverter.ToInt32(dataLenBytes, 0);
                }
                if (datalen > 0)
                {
                    //读取字节数据
                    data = persistence.Read(per.TagKey, true, datalen);
                    //增加当前读取长度
                    curReadCount += data.Length;
                    if (data.Length > 2 && data[0] == 0x7e && data[data.Length - 1] == 0x7e)
                    {
                        //增加记数器 9,223,372,036,854,775,808 
                        per.SendCount++;
                        //增加读取的有效数据长度
                        datavalidLen += data.Length;
                        //增加可以读取的数据
                        sendArrs.Add(data);
                    }
                    else
                    {

                        inCom.Logs.LogProvider.Create().Write(inCom.Logs.EventType.Warning, "读取持久化实时数据发送出错..." + per.TagKey);
                        int k = 0;
                        //最多读取100万次
                        //数据出错则将数据读取到最后的字节
                        while (persistence.Read(per.TagKey, true).Length > 0)
                        {
                            k++;
                            if (k > 100 * 10000)
                            {
                                return;
                            }
                        }
                    }
                }
                else
                {

                    //完成数据
                    CompleteCycle(per);
                    System.Threading.Thread.Sleep(20);

                    break;
                }
                #endregion
                //当读取最大值时直接跳过
                if (curReadCount >= maxReadlen)
                {
                    break;
                }
            }
            //使用此方式单个节点发送数据可以破万
            if (datavalidLen > 0)
            {
                #region 构造发送包并发送数据
                //将要发送的数据包
                byte[] willSendbytes = new byte[datavalidLen];

                int perIndex = 0;
                foreach (var arr in sendArrs)
                {
                    Buffer.BlockCopy(arr, 0, willSendbytes, perIndex, arr.Length);
                    perIndex += arr.Length;
                }
                //需要补发数据
                if (!SendNew)
                {
                    //这里是发送代码
                    InCom.SyncStatus.SyncPointManager.Instance.Send(per.SyncPoint, willSendbytes, false);
                }
                else
                {
                    //发送最新数据
                    InCom.SyncStatus.SyncPointManager.Instance.Send(per.SyncPoint, willSendbytes, false);
                }
                System.Threading.Interlocked.Add(ref RealDataTransmissionSendBuffer.Instance.sendCount, sendArrs.Count);
                #endregion
            }
            //没有读取到数据
            if (datavalidLen == 0)
            {
                //读取一包数据
               var sendBytesPackaeg = ReadPack(per);
                if(sendBytesPackaeg!=null)
                {
                    //这里是发送代码
                    InCom.SyncStatus.SyncPointManager.Instance.Send(per.SyncPoint, sendBytesPackaeg, false);
                    System.Threading.Interlocked.Add(ref RealDataTransmissionSendBuffer.Instance.sendCount,1);
                }
            }






        }
        /// <summary>
        /// 读取单包数据
        /// </summary>
        /// <returns></returns>
        protected byte[] ReadPack(SendTagsBuffer per)
        {
            byte[] data = null;

            int datalen = 0;
            //正常可以使用节点
            var dataLenBytes = persistence.Read(per.TagKey, true, 4);
            if (dataLenBytes.Length == 4)
            {
                datalen = BitConverter.ToInt32(dataLenBytes, 0);
            }
            if (datalen > 0)
            {
                //读取字节数据
                data = persistence.Read(per.TagKey, true, datalen);
                if (data.Length > 2 && data[0] == 0x7e && data[data.Length - 1] == 0x7e)
                {
                    //增加记数器 9,223,372,036,854,775,808 
                    per.SendCount++;
                    return data;
                }
                else
                {
                    inCom.Logs.LogProvider.Create().Write(inCom.Logs.EventType.Warning, "读取持久化实时数据发送出错..." + per.TagKey);
                    int k = 0;
                    //最多读取100万次
                    //数据出错则将数据读取到最后的字节
                    while (persistence.Read(per.TagKey, true).Length > 0)
                    {
                        k++;
                        if (k > 100 * 10000)
                        {
                            return null;
                        }
                    }
                }
            }
            else
            {
                //读取完成数据
                CompleteCycle(per);
                System.Threading.Thread.Sleep(20);
            }

            return null;
        }

        /// <summary>
        /// 完成循环
        /// </summary>
        /// <param name="per">状态变量</param>
        private void CompleteCycle(SendTagsBuffer per)
        {
            var obj = persistence.GetPersistenceStatus(per.TagKey);
            if (obj == null || obj.FileReadWrite == null)
            {
                return;
            }
            //5秒没有任何数据写入
           // if (obj.FileReadWrite.CureentWrite > 0 && obj.CurrentIndex == obj.FileReadWrite.CureentWrite && obj.FileReadWrite.LastSaveTime.AddSeconds(5) < DateTime.Now)
            if (obj.FileReadWrite.CureentWrite > 0 && obj.CurrentIndex == obj.FileReadWrite.CureentWrite && obj.FileReadWrite.LastSaveTime.AddMinutes(10) < DateTime.Now)
            {
                if (!per.CanRun)
                {
                    return;
                }

                //设置运行完成
                per.CanRun = false;
                per.Task.CanToKen.Cancel();
                double ss = (DateTime.Now - per.StartTime).TotalSeconds;
                Console.WriteLine(per.TagKey + "发送.." + per.SendCount + "完成 用时:" + ss.ToString("f2") + "  速度" + per.SendCount / ss + "/s");
                //设置对像操作完成等待持久化回收
                obj.IsComplated = true;
                //释放资源
                obj.FileReadWrite.Dispose();
                obj.FileReadWrite = null;
                //删除集合对像
                if (Sendtasks.TryRemove(per.TagKey, out per))
                {
                    Console.WriteLine("将线程移除队列,发送总数" + per.SendCount);
                    return;
                }
            }
        }

        #endregion




    }


    /// <summary>
    /// 发送到缓冲区的buffer
    /// </summary>
    public class SendTagsBuffer
    {
        #region 读取队列总数
        /// <summary>
        /// 获取贴总数
        /// </summary>
        public int GetFrameCount
        {
            get { return bytesqueue.Count; }
        }

        /// <summary>
        /// 数据队列 
        /// </summary>
        protected Queue<List<byte>> bytesqueue = new Queue<List<byte>>();

        /// <summary>
        /// 数据集合
        /// </summary>
        private List<byte> bytes = new List<byte>();

        /// <summary>
        /// 增加需要发送的数据
        /// </summary>
        /// <param name="data">当前数据</param>
        public void Append(byte[] data)
        {
            for (int i = 0; i < data.Length; i++)
            {
                //找协议头,尾
                if (data[i] == 0x7e)
                {
                    bytes.Add(data[i]);
                    //如果协议包含有效数据内容
                    if (bytes.Count > 1)
                    {
                        lock (bytes)
                        {
                            var list = bytes;
                            bytes = new List<byte>();
                            if (list[0] == 0x7e && list[list.Count - 1] == 0x7e)
                            {
                                //添加到集合对像
                                bytesqueue.Enqueue(list);
                            }
                            else
                            {
                                var 出错 = "";
                                inCom.Logs.LogProvider.Create().Write(inCom.Logs.EventType.Warning, "读取持久化实时数据发送出错...");
                                //出错
                            }

                        }
                    }
                }
                else
                {
                    bytes.Add(data[i]);
                }

            }
        }

        /// <summary>
        /// 得到一贴数据,没有数据则返回null
        /// </summary>
        /// <returns></returns>
        public List<byte> GetFarme()
        {

            lock (bytesqueue)
            {
                return bytesqueue.Count > 0 ? bytesqueue.Dequeue() : null;
            }
        }
        #endregion



        #region 状态数据
        /// <summary>
        /// 开始时间
        /// </summary>
        public DateTime StartTime { get; set; }

        /// <summary>
        /// 发送包总数
        /// </summary>
        public long SendCount { get; set; }

        /// <summary>
        /// 写入包
        /// </summary>
        public long WriteCount { get; set; }

        /// <summary>
        /// 数据数据对像
        /// </summary>
        public object UserToKen { get; set; }

        /// <summary>
        /// 标识
        /// </summary>
        public string TagKey { get; set; }

        /// <summary>
        /// 对应的任务对像
        /// </summary>
        public TaskEx Task { get; set; }


        /// <summary>
        /// 数据同步节点类型
        /// </summary>
        public SyncPoint SyncPoint { get; set; }

        /// <summary>
        /// 可以运行
        /// </summary>
        public bool CanRun { get; set; }


        /// <summary>
        /// 持久化数据状态
        /// </summary>
        public PersistenceStatus PerStatus { get; set; }

        #endregion

    }
}
